package io.gitee.zicai.stream.receive

import io.gitee.zicai.stream.processor.MsgDelayProcessor
import io.gitee.zicai.stream.processor.MsgProcessor
import io.gitee.zicai.stream.util.LogKit
import org.springframework.amqp.rabbit.annotation.RabbitHandler
import org.springframework.amqp.rabbit.annotation.RabbitListener
import org.springframework.cloud.stream.annotation.EnableBinding
import org.springframework.cloud.stream.annotation.StreamListener
import org.springframework.integration.annotation.Poller
import org.springframework.integration.annotation.ServiceActivator
import org.springframework.messaging.handler.annotation.Header
import org.springframework.messaging.handler.annotation.Payload
import org.springframework.messaging.support.ErrorMessage

/**
 * MsgReceiver
 * @author zicai
 * @since 2020-05-11
 */
@EnableBinding(value = [ MsgProcessor::class, MsgDelayProcessor::class ])
open class MsgReceiver {

    @StreamListener(MsgProcessor.INPUT, condition = "headers['type'] == 'even'")
    open fun receiveEven(@Payload msg: Int) {
        println("${LogKit.nowTime()} - Even payload -> $msg")
    }

    @StreamListener(MsgProcessor.INPUT, condition = "headers['type'] == 'odd'")
    open fun receiveOdd(@Payload msg: Int) {
        println("${LogKit.nowTime()} - Odd payload -> $msg")
    }

    @StreamListener(MsgDelayProcessor.INPUT, condition = "headers['type'] == 'ok'")
    open fun receiveDelay(@Payload msg: Int) {
        println("${LogKit.nowTime()} - Delay payload -> $msg")
    }

    @StreamListener(MsgDelayProcessor.INPUT, condition = "headers['type'] == 'error'")
    open fun receiveDelayError(@Payload msg: Int) {
        println("${LogKit.nowTime()} - Delay payload error -> $msg")
        throw RuntimeException("error msg -> $msg")
    }

    @ServiceActivator(inputChannel = "msg-delay-queue.msg-delay-queue-consumer.errors", poller = [
        Poller(fixedDelay = "1000", maxMessagesPerPoll = "1")
    ])
    open fun receiveDelayErrorHandler(msg: ErrorMessage) {
        println("${LogKit.nowTime()} - Delay error -> ${msg.payload}")
    }

    @RabbitListener(queues = ["amqp-test-queue"])
    @RabbitHandler
    open fun amqpListener(@Payload msg: String, @Header type: String) {
        println("${LogKit.nowTime()} - amqp receive type -> $type, msg -> $msg")
    }
}